package com.util;

import org.apache.commons.io.FileUtils;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.nio.charset.Charset;
import java.util.concurrent.Future;

public class KafkaDataSendTaskThread implements Runnable{

    private static final Logger logger = LoggerFactory.getLogger(KafkaDataSendTaskThread.class);

    private String code;
    private String filePath;
    private String headLine;
    private String fileEncoding;
    private KafkaProducerPool kafkaProducerPool;
    private String kafkaTopic;
    private String rsyncTopic;


    public KafkaDataSendTaskThread() {
    }

    public KafkaDataSendTaskThread(String code, String filePath, String headLine, String fileEncoding, KafkaProducerPool kafkaProducerPool, String kafkaTopic, String rsyncTopic) {
        this.code = code;
        this.filePath = filePath;
        this.headLine = headLine;
        this.fileEncoding = fileEncoding;
        this.kafkaProducerPool = kafkaProducerPool;
        this.kafkaTopic = kafkaTopic;
        this.rsyncTopic = rsyncTopic;
    }

    @Override
    public void run() {

        KafkaProducer kafkaProducer = kafkaProducerPool.getKafkaProducer();

        try( BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(filePath), Charset.forName(fileEncoding))) ){

            String fileHead = bufferedReader.readLine();
            String headStr = (null == headLine || "".equals(headLine)) ? fileHead : headLine;
            String fileNameStr = new File(filePath).getName();

            String lineStr = "";
            while((lineStr = bufferedReader.readLine()) != null){
                String data = code + "@@" + fileNameStr + "@@" + headStr + "@@" + lineStr;
                Future send = kafkaProducer.send(new ProducerRecord(kafkaTopic, data, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            System.out.println("异常 失败 " + exception);
                        }
                    }
                }));
            }

            // 移动文件到done目录
            FileUtils.moveFile(new File(filePath), new File(filePath.replace("move", "done")));
            kafkaProducer.send(new ProducerRecord(rsyncTopic, code + "@@" + fileNameStr + "@@" + filePath));

            kafkaProducerPool.returnKafkaProducer(kafkaProducer);

        } catch (FileNotFoundException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (Throwable t){
            logger.error("", t);
        }


    }


}
